package com.hzgj.bcl.soa.merge;

import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.hzgj.bcl.redis.RedisClient;
import com.hzgj.bcl.stats.RedisKey;
import com.hzgj.bcl.util.jmx.annotation.MBean;
import com.hzgj.soa.spring.ServerBean;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.Calendar;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import static com.hzgj.bcl.util.lang.ThreadPoolKit.getScheduledExecutorService;

/**
 * Created by Administrator on 2016/8/11.
 */
@Component
@MBean(description = "CliMergeBootstrap")
public class Bootstrap implements MergeControl {

    private Logger logger = LoggerFactory.getLogger(getClass());

    public static void main(String[] args) throws Exception {
        new ClassPathXmlApplicationContext("classpath:config.xml").getBean(ServerBean.class);
    }

    @Autowired
    private RedisClient redisClient;
    private ScheduledExecutorService scheduledExecutorService = getScheduledExecutorService("merge redis data");
    private ThreadPoolExecutor threadPoolExecutor;
    private Map<String, Merge> mergeCache = Maps.newHashMap();
    private AtomicBoolean run = new AtomicBoolean(true);

    public Cache<String, String> cache = CacheBuilder.newBuilder().expireAfterWrite(2, TimeUnit.MINUTES).build();


    @PostConstruct
    public void start() {

        final Set<String> members = Sets.newHashSet();
        scheduledExecutorService.scheduleWithFixedDelay(() -> {
            try {
                if (run.get()) {
                    members.addAll(redisClient.smembers(RedisKey.SERVICE_METHOD_KEY));
                    Set<String> removing = Sets.difference(mergeCache.keySet(), members);
                    removing.forEach(s -> {
                        Merge merge = mergeCache.remove(s);
                        merge.stop();
                        logger.info("stop run {} ", s);
                    });
                    long now = System.currentTimeMillis();
                    Calendar calendar = Calendar.getInstance();
                    calendar.add(Calendar.DAY_OF_YEAR, -30);
                    for (String key : members) {
                        String value = cache.getIfPresent(key);
                        if (value != null) {
                            logger.info("{} is handling", key);
                            continue;
                        }
                        if (logger.isDebugEnabled()) {
                            logger.debug("start to run {}", key);
                        }
                        threadPoolExecutor.execute(() -> {
                            try {
                                Merge merge = mergeCache.computeIfAbsent(key, s -> new Merge(key, redisClient));
                                merge.doIt(calendar.getTimeInMillis(), now);
                            } catch (Exception e) {
                                logger.error("{}", e);
                            } finally {
                                cache.invalidate(key);
                            }
                        });
                        cache.put(key, key);
                    }
                } else {
                    logger.info("scheduledExecutorService has been stopped.");
                }
            } catch (Exception e) {
                logger.error("{}", e);
            }
        }, 0, 2, TimeUnit.SECONDS);

        threadPoolExecutor = new ThreadPoolExecutor(2, Runtime.getRuntime().availableProcessors() * 2 + 1, 3,
                TimeUnit.SECONDS, new ArrayBlockingQueue<>(100),
                new ThreadPoolExecutor.CallerRunsPolicy());
    }


    @Override
    public MergeControlResult drainMerge() {
        if (run.compareAndSet(true, false)) {
            while (threadPoolExecutor.getTaskCount() > 0) {
            }
            logger.info("start to drain merge.");
            mergeCache.values().forEach(merge -> {
                merge.stop();
            });
            threadPoolExecutor.shutdown();
            logger.info("drain over");
        } else {
            return MergeControlResult.builder().success(false).detail("scheduledExecutorService has been stopped").build();
        }

        return MergeControlResult.builder().success(true).build();
    }

    @Override
    public ExecutorInfo getExecutorTask() {
        return ExecutorInfo.builder().count(threadPoolExecutor.getTaskCount()).
                status(threadPoolExecutor.isShutdown() ? "shutdown" : threadPoolExecutor.isTerminated() ? "terminated" : "").build();
    }

}
